/*
 * Singularity Engine
 * Copyright (C) 2010 Vyacheslav Vorobyov, All rights reserved.
 * See LICENSE for full license information.
 */

/**
 * 
 */
package org.singularity.mapred.framework;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;

import org.singularity.io.AbstractTransitPipe;
import org.singularity.io.CloseablePipe;

/**
 * @author vjache
 *
 */
public class RunnablePipe<VT> extends AbstractTransitPipe<VT, VT> implements Runnable {

	private final UUID _id = UUID.randomUUID();
	private final ConcurrentLinkedQueue<VT> _queue = new ConcurrentLinkedQueue<VT>();
	private final Executor _exec;
	private long _cnt = 0;
	

	public RunnablePipe(CloseablePipe<VT> aNextOne, Executor aExec) {
		super(aNextOne);
		_exec = aExec;
	}
	
	/* (non-Javadoc)
	 * @see java.lang.Runnable#run()
	 */
	@Override
	public void run() {
		VT item;
		while((item = _queue.poll())!=null)
		{
			try {
				_next.push(item);
			} catch (IOException e) {
				throw new RuntimeException(e);
			}
		}
	}
	
	@Override
	public void flush() throws IOException {
		System.out.println("FLUSH");
		flush_();
		while(!_queue.isEmpty())
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				throw new IOException(e);
			}
		super.flush();
	}

	private void flush_() {
		_exec.execute(this);
		_cnt = 0;
	}

	@Override
	public void push(VT aItem) throws IOException {
		_queue.add(aItem);
		_cnt++;
		if(_cnt>=100)
			flush_();
	}

	@Override
	public int hashCode() {
		return _id.hashCode();
	}
}
